
#include "stdafx.h"
#include "IPCCallMgr.h"
#include "IPCCallPacket.h"

IPCCallMgr::IPCCallMgr()
:pProcessStoreReqThreadHandlerContext_(),pProcessFetchReqThreadHandlerContext_(),pProcessAppAttachReqThreadHandlerContext_()
{

}

IPCCallMgr::~IPCCallMgr()
{
	pProcessStoreReqThreadHandlerContext_.reset();
	pProcessFetchReqThreadHandlerContext_.reset();
	pProcessAppAttachReqThreadHandlerContext_.reset();
}

retT IPCCallMgr::init()
{
	R(asyframe::HandlerBase::init());
	pProcessStoreReqThreadHandlerContext_.reset(new asyframe::StaticThreadHandlerContext());
	pProcessFetchReqThreadHandlerContext_.reset(new asyframe::StaticThreadHandlerContext());
	pProcessAppAttachReqThreadHandlerContext_.reset(new asyframe::StaticThreadHandlerContext());
	RETSUCCESS;
}

retT IPCCallMgr::start()
{
	R(asyframe::HandlerBase::start());
	R(asyframe::StaticThreadHandler::justStartThread(boost::bind(&IPCCallMgr::processStoreReq, boost::dynamic_pointer_cast<IPCCallMgr>(shared_from_this())), pProcessStoreReqThreadHandlerContext_));
	R(asyframe::StaticThreadHandler::justStartThread(boost::bind(&IPCCallMgr::processFetchReq, boost::dynamic_pointer_cast<IPCCallMgr>(shared_from_this())), pProcessFetchReqThreadHandlerContext_));
	R(asyframe::StaticThreadHandler::justStartThread(boost::bind(&IPCCallMgr::processAppAttachReq, boost::dynamic_pointer_cast<IPCCallMgr>(shared_from_this())), pProcessAppAttachReqThreadHandlerContext_));
	RETSUCCESS;
}

retT IPCCallMgr::close()
{
	R(asyframe::HandlerBase::close());
	if(pProcessStoreReqThreadHandlerContext_ && pProcessStoreReqThreadHandlerContext_->pThread_)
	{
		pProcessStoreReqThreadHandlerContext_->pThread_->interrupt();
		pProcessStoreReqThreadHandlerContext_->pThread_.reset();
	}
	if(pProcessFetchReqThreadHandlerContext_ && pProcessFetchReqThreadHandlerContext_->pThread_)
	{
		pProcessFetchReqThreadHandlerContext_->pThread_->interrupt();
		pProcessFetchReqThreadHandlerContext_->pThread_.reset();
	}
	if(pProcessAppAttachReqThreadHandlerContext_ && pProcessAppAttachReqThreadHandlerContext_->pThread_)
	{
		pProcessAppAttachReqThreadHandlerContext_->pThread_->interrupt();
		pProcessAppAttachReqThreadHandlerContext_->pThread_.reset();
	}
	RETSUCCESS;
}

retT IPCCallMgr::processStoreReq()
{
	try
	{
		boost::interprocess::message_queue::remove(sendStoreReqMessageQueueName_.c_str());
		boost::interprocess::message_queue mq(boost::interprocess::create_only ,sendStoreReqMessageQueueName_.c_str(), maxMessageNumber, maxMessageSize);

		unsigned int priority;
		std::size_t recvdSize;
		std::vector<asyframe::byte> dataBuffer(maxMessageSize);
		boost::shared_ptr<asyframe::IoBuffer> pBuffer(new asyframe::IoBuffer(&dataBuffer));

		while(true)
		{
			pBuffer->clear();
			mq.receive(pBuffer->data(), pBuffer->capacity(), recvdSize, priority);
			_R(pBuffer->produce(recvdSize))
			{
				if(ret.status_ != CONST_COMMON_SUCCESS)
					continue;
			}R_
			boost::shared_ptr<SendStoreReqIPCCallPacket> pSendStoreReqIPCCallPacket(new SendStoreReqIPCCallPacket());
			_R(pSendStoreReqIPCCallPacket->readFromBuffer(pBuffer))
			{
				if(ret.status_ != CONST_COMMON_SUCCESS)
					continue;
			}R_
			if(pSendStoreReqIPCCallPacket->pReloadPacket_ && pSendStoreReqIPCCallPacket->pReloadPacket_->msgContents_.messageBody_.elements_.size() > 0)
			{
				boost::shared_ptr<reload::StoreReq> pStoreReq = boost::dynamic_pointer_cast<reload::StoreReq>(pSendStoreReqIPCCallPacket->pReloadPacket_->msgContents_.messageBody_.elements_[0]);
				if(pStoreReq)
				{
					_R(reload::sendStoreReq(pSendStoreReqIPCCallPacket->overlayInstanceName_.toString(), pSendStoreReqIPCCallPacket->resourceName_.toString().c_str(),
						pStoreReq->kindData_, boost::bind(&IPCCallMgr::onSendStoreReq,
						boost::dynamic_pointer_cast<IPCCallMgr>(shared_from_this()), pSendStoreReqIPCCallPacket, _1, _2)))
					{
						if(ret.status_ != CONST_COMMON_SUCCESS)
							continue;
					}R_
				}
			}
		}
	}
	catch(boost::interprocess::interprocess_exception &ex)
	{
		boost::interprocess::message_queue::remove(sendStoreReqMessageQueueName_.c_str());
		LOG_INFO(boost::format( "interprocess exception : %1% in %2% %3% %4%" ) % ex.what() % __FUNCTION__ % __FILE__ % __LINE__);
		RET(ERROR_IPCCALLMGR_INTERPROCESS_EXCEPTION);
	}
	boost::interprocess::message_queue::remove(sendStoreReqMessageQueueName_.c_str());
	RETSUCCESS;
}

retT IPCCallMgr::processFetchReq()
{
	try
	{
		boost::interprocess::message_queue::remove(sendFetchReqMessageQueueName_.c_str());
		boost::interprocess::message_queue mq(boost::interprocess::create_only ,sendFetchReqMessageQueueName_.c_str(), maxMessageNumber, maxMessageSize);

		unsigned int priority;
		std::size_t recvdSize;
		std::vector<asyframe::byte> dataBuffer(maxMessageSize);
		boost::shared_ptr<asyframe::IoBuffer> pBuffer(new asyframe::IoBuffer(&dataBuffer));

		while(true)
		{
			pBuffer->clear();
			mq.receive(pBuffer->data(), pBuffer->capacity(), recvdSize, priority);
			_R(pBuffer->produce(recvdSize))
			{
				if(ret.status_ != CONST_COMMON_SUCCESS)
					continue;
			}R_
			boost::shared_ptr<SendFetchReqIPCCallPacket> pSendFetchReqIPCCallPacket(new SendFetchReqIPCCallPacket());
			_R(pSendFetchReqIPCCallPacket->readFromBuffer(pBuffer))
			{
				if(ret.status_ != CONST_COMMON_SUCCESS)
					continue;
			}R_
			if(pSendFetchReqIPCCallPacket->pReloadPacket_ && pSendFetchReqIPCCallPacket->pReloadPacket_->msgContents_.messageBody_.elements_.size() > 0)
			{
				boost::shared_ptr<reload::FetchReq> pFetchReq = boost::dynamic_pointer_cast<reload::FetchReq>(pSendFetchReqIPCCallPacket->pReloadPacket_->msgContents_.messageBody_.elements_[0]);
				if(pFetchReq)
				{
					_R(reload::sendFetchReq(pSendFetchReqIPCCallPacket->overlayInstanceName_.toString(), pSendFetchReqIPCCallPacket->resourceName_.toString().c_str(),
						pFetchReq->specifiers_, boost::bind(&IPCCallMgr::onSendFetchReq,
						boost::dynamic_pointer_cast<IPCCallMgr>(shared_from_this()), pSendFetchReqIPCCallPacket, _1, _2)))
					{
						if(ret.status_ != CONST_COMMON_SUCCESS)
							continue;
					}R_
				}
			}
		}
	}
	catch(boost::interprocess::interprocess_exception &ex)
	{
		boost::interprocess::message_queue::remove(sendFetchReqMessageQueueName_.c_str());
		LOG_INFO(boost::format( "interprocess exception : %1% in %2% %3% %4%" ) % ex.what() % __FUNCTION__ % __FILE__ % __LINE__);
		RET(ERROR_IPCCALLMGR_INTERPROCESS_EXCEPTION);
	}
	boost::interprocess::message_queue::remove(sendFetchReqMessageQueueName_.c_str());
	RETSUCCESS;
}

retT IPCCallMgr::processAppAttachReq()
{
	try
	{
		boost::interprocess::message_queue::remove(sendAppAttachReqMessageQueueName_.c_str());
		boost::interprocess::message_queue mq(boost::interprocess::create_only ,sendAppAttachReqMessageQueueName_.c_str(), maxMessageNumber, maxMessageSize);

		unsigned int priority;
		std::size_t recvdSize;
		std::vector<asyframe::byte> dataBuffer(maxMessageSize);
		boost::shared_ptr<asyframe::IoBuffer> pBuffer(new asyframe::IoBuffer(&dataBuffer));

		while(true)
		{
			pBuffer->clear();
			mq.receive(pBuffer->data(), pBuffer->capacity(), recvdSize, priority);
			_R(pBuffer->produce(recvdSize))
			{
				if(ret.status_ != CONST_COMMON_SUCCESS)
					continue;
			}R_
			boost::shared_ptr<IPCCallPacketBase> pIPCCallPacketBase(new IPCCallPacketBase());
			_R(pIPCCallPacketBase->readFromBuffer(pBuffer))
			{
				if(ret.status_ != CONST_COMMON_SUCCESS)
					continue;
			}R_
			_R(reload::sendAppAttachReq(pIPCCallPacketBase->overlayInstanceName_.toString(), pIPCCallPacketBase->resourceName_.toString().c_str(),
				boost::bind(&IPCCallMgr::onSendAppAttachReq,
				boost::dynamic_pointer_cast<IPCCallMgr>(shared_from_this()), pIPCCallPacketBase, _1, _2)))
			{
				if(ret.status_ != CONST_COMMON_SUCCESS)
					continue;
			}R_
		}
	}
	catch(boost::interprocess::interprocess_exception &ex)
	{
		boost::interprocess::message_queue::remove(sendAppAttachReqMessageQueueName_.c_str());
		LOG_INFO(boost::format( "interprocess exception : %1% in %2% %3% %4%" ) % ex.what() % __FUNCTION__ % __FILE__ % __LINE__);
		RET(ERROR_IPCCALLMGR_INTERPROCESS_EXCEPTION);
	}
	boost::interprocess::message_queue::remove(sendAppAttachReqMessageQueueName_.c_str());
	RETSUCCESS;
}

retT IPCCallMgr::onSendStoreReq( boost::shared_ptr<SendStoreReqIPCCallPacket> _pSendStoreReqIPCCallPacket, reload::TransactionStatus _transactionStatus, boost::shared_ptr<reload::ReloadNetworkDataContext> _pReloadNetworkDataContext )
{
	try
	{
		if(_pSendStoreReqIPCCallPacket)
		{
			boost::interprocess::message_queue mq(boost::interprocess::open_only, _pSendStoreReqIPCCallPacket->messageQueueName_.toString().c_str());
			IPCCallResultBase iPCCallResultBase;
			if(_transactionStatus.status_ == reload::TransactionStatus_Success)
			{
				if(_pReloadNetworkDataContext && _pReloadNetworkDataContext->pReloadPacket_ && _pReloadNetworkDataContext->pReloadPacket_->msgContents_.messageCode_ != reload::MessageCode_Error)
				{
					iPCCallResultBase.result_ =0; 
				}
				else
				{
					if(_pReloadNetworkDataContext && _pReloadNetworkDataContext->pReloadPacket_ && _pReloadNetworkDataContext->pReloadPacket_->msgContents_.messageCode_ == reload::MessageCode_Error)
					{
						iPCCallResultBase.result_ = _pReloadNetworkDataContext->pReloadPacket_->msgContents_.errorResponse_.errorCode_;
					}
					else
					{
						iPCCallResultBase.result_ = -1;
					}
				}
			}
			else
			{
				iPCCallResultBase.result_ = -1;
			}
			std::vector<asyframe::byte> dataBuffer(iPCCallResultBase.size());
			boost::shared_ptr<asyframe::IoBuffer> pBuffer(new asyframe::IoBuffer(&dataBuffer));
			pBuffer->clear();
			R(iPCCallResultBase.writeToBuffer(pBuffer));
			boost::posix_time::ptime currentTime = boost::posix_time::second_clock::local_time();
			boost::posix_time::ptime endTime = currentTime + boost::posix_time::seconds(10);
			if(!mq.timed_send(pBuffer->data(), iPCCallResultBase.size(), 0, endTime))
			{
				LOG_INFO(boost::format( "message is full, can't send message timely! in %1% %2% %3%" ) % __FUNCTION__ % __FILE__ % __LINE__);
			}
		}
	}
	catch(boost::interprocess::interprocess_exception &ex)
	{
		LOG_INFO(boost::format( "interprocess exception : %1% in %2% %3% %4%" ) % ex.what() % __FUNCTION__ % __FILE__ % __LINE__);
		RET(INFO_IPCCALLMGR_INTERPROCESS_EXCEPTION);
	}
	RETSUCCESS;
}

retT IPCCallMgr::onSendFetchReq( boost::shared_ptr<SendFetchReqIPCCallPacket> _pSendFetchReqIPCCallPacket, reload::TransactionStatus _transactionStatus, boost::shared_ptr<reload::ReloadNetworkDataContext> _pReloadNetworkDataContext )
{
	try
	{
		if(_pSendFetchReqIPCCallPacket)
		{
			boost::interprocess::message_queue mq(boost::interprocess::open_only, _pSendFetchReqIPCCallPacket->messageQueueName_.toString().c_str());
			SendFetchReqIPCCallResult sendFetchReqIPCCallResult;
			if(_transactionStatus.status_ == reload::TransactionStatus_Success)
			{
				if(_pReloadNetworkDataContext && _pReloadNetworkDataContext->pReloadPacket_ && _pReloadNetworkDataContext->pReloadPacket_->msgContents_.messageCode_ != reload::MessageCode_Error)
				{
					sendFetchReqIPCCallResult.result_ = 0; 
					sendFetchReqIPCCallResult.pReloadPacket_ = _pReloadNetworkDataContext->pReloadPacket_;
				}
				else
				{
					if(_pReloadNetworkDataContext && _pReloadNetworkDataContext->pReloadPacket_ && _pReloadNetworkDataContext->pReloadPacket_->msgContents_.messageCode_ == reload::MessageCode_Error)
					{
						sendFetchReqIPCCallResult.result_ = _pReloadNetworkDataContext->pReloadPacket_->msgContents_.errorResponse_.errorCode_;
					}
					else
					{
						sendFetchReqIPCCallResult.result_ = -1;
					}
				}
			}
			else
			{
				sendFetchReqIPCCallResult.result_ = -1;
			}
			std::vector<asyframe::byte> dataBuffer(sendFetchReqIPCCallResult.size());
			boost::shared_ptr<asyframe::IoBuffer> pBuffer(new asyframe::IoBuffer(&dataBuffer));
			pBuffer->clear();
			R(sendFetchReqIPCCallResult.writeToBuffer(pBuffer));
			boost::posix_time::ptime currentTime = boost::posix_time::second_clock::local_time();
			boost::posix_time::ptime endTime = currentTime + boost::posix_time::seconds(10);
			if(!mq.timed_send(pBuffer->data(), sendFetchReqIPCCallResult.size(), 0, endTime))
			{
				LOG_INFO(boost::format( "message is full, can't send message timely! in %1% %2% %3%" ) % __FUNCTION__ % __FILE__ % __LINE__);
			}
		}
	}
	catch(boost::interprocess::interprocess_exception &ex)
	{
		LOG_INFO(boost::format( "interprocess exception : %1% in %2% %3% %4%" ) % ex.what() % __FUNCTION__ % __FILE__ % __LINE__);
		RET(INFO_IPCCALLMGR_INTERPROCESS_EXCEPTION);
	}
	RETSUCCESS;
}

retT IPCCallMgr::onSendAppAttachReq( boost::shared_ptr<IPCCallPacketBase> _pIPCCallPacketBase, reload::TransactionStatus _transactionStatus, boost::shared_ptr<reload::ReloadNetworkDataContext> _pReloadNetworkDataContext )
{
	try
	{
		if(_pIPCCallPacketBase)
		{
			boost::interprocess::message_queue mq(boost::interprocess::open_only, _pIPCCallPacketBase->messageQueueName_.toString().c_str());
			SendAppAttachReqIPCCallResult sendAppAttachReqIPCCallResult;
			if(_transactionStatus.status_ == reload::TransactionStatus_Success)
			{
				if(_pReloadNetworkDataContext && _pReloadNetworkDataContext->pReloadPacket_ && _pReloadNetworkDataContext->pReloadPacket_->msgContents_.messageCode_ != reload::MessageCode_Error)
				{
					sendAppAttachReqIPCCallResult.result_ =0; 
					if(_pReloadNetworkDataContext && _pReloadNetworkDataContext->pReloadPacket_ && _pReloadNetworkDataContext->pReloadPacket_->msgContents_.messageBody_.elements_.size() > 0)
					{
						if(boost::shared_ptr<reload::AppAttachAns> pAppAttachAns = boost::dynamic_pointer_cast<reload::AppAttachAns>(_pReloadNetworkDataContext->pReloadPacket_->msgContents_.messageBody_.elements_[0]))
						{
							for(std::size_t i = 0; i < pAppAttachAns->candidates_.array_.size(); ++i)
							{
								sendAppAttachReqIPCCallResult.strIP_.setValue(pAppAttachAns->candidates_.array_[i].addrPort_.addr_.to_string());
							}
						}
					}
				}
				else
				{
					if(_pReloadNetworkDataContext && _pReloadNetworkDataContext->pReloadPacket_ && _pReloadNetworkDataContext->pReloadPacket_->msgContents_.messageCode_ == reload::MessageCode_Error)
					{
						sendAppAttachReqIPCCallResult.result_ = _pReloadNetworkDataContext->pReloadPacket_->msgContents_.errorResponse_.errorCode_;
					}
					else
					{
						sendAppAttachReqIPCCallResult.result_ = -1;
					}
				}
			}
			else
			{
				sendAppAttachReqIPCCallResult.result_ = -1;
			}
			std::vector<asyframe::byte> dataBuffer(sendAppAttachReqIPCCallResult.size());
			boost::shared_ptr<asyframe::IoBuffer> pBuffer(new asyframe::IoBuffer(&dataBuffer));
			pBuffer->clear();
			R(sendAppAttachReqIPCCallResult.writeToBuffer(pBuffer));
			boost::posix_time::ptime currentTime = boost::posix_time::second_clock::local_time();
			boost::posix_time::ptime endTime = currentTime + boost::posix_time::seconds(10);
			if(!mq.timed_send(pBuffer->data(), sendAppAttachReqIPCCallResult.size(), 0, endTime))
			{
				LOG_INFO(boost::format( "message is full, can't send message timely! in %1% %2% %3%" ) % __FUNCTION__ % __FILE__ % __LINE__);
			}
		}
	}
	catch(boost::interprocess::interprocess_exception &ex)
	{
		LOG_INFO(boost::format( "interprocess exception : %1% in %2% %3% %4%" ) % ex.what() % __FUNCTION__ % __FILE__ % __LINE__);
		RET(INFO_IPCCALLMGR_INTERPROCESS_EXCEPTION);
	}
	RETSUCCESS;
}
